Usage documentation
Eizen SDK Usage Guide
1. Introduction
The Eizen SDK provides a convenient Python interface for interacting with the Eizen platform's API. It allows you to manage tenants, analytics configurations, sources, models, and trigger various processing tasks like model training, inference, and video analysis.
Prerequisites
Before installing the SDK, make sure you have the following:
- Python 3.8 or above installed on your system
pip(Python package manager) available- Access to the Eizen platform with a valid refresh token
2. Installing the SDK
Run the following command to install the latest version of the Eizen SDK:
pip install eizen_sdk
*(Note: If eizen\_sdk is not the correct package name on PyPI, use the actual published name).*
## **3. Authentication & Initialization**
To use the SDK, you need to authenticate with the Eizen platform. The primary method uses an access\_token and an api\_key.
For Eizen Environment:
import os
from dotenv import load\_dotenv
from eizen_sdk import EizenSDK
REFRESH_TOKEN = "PASTE_YOUR_REFRESH_TOKEN_HERE"
try:
sdk = EizenSDK(refresh_token=REFRESH_TOKEN)
print("SDK initialized successfully.")
except Exception as e:
print(f"Error initializing SDK: {e}")
For On-prem System:
import os
from dotenv import load\_dotenv
from eizen\_sdk import EizenSDK
\# Load environment variables from .env file
load\_dotenv()
sdk = EizenSDK(
` `refresh\_token=os.getenv('refresh\_token'),
` `base\_url=os.getenv('base\_url'),
` `keycloak\_base\_url=os.getenv('keycloak\_base\_url'),
` `client\_id=os.getenv('client\_id'),
` `gateway\_base\_url=os.getenv('gateway\_base\_url')
)
print("Initilized successful")
## **Verifying the Installation**
**To confirm that the SDK works, try listing available tenants**
try:
tenants = sdk.list_tenants()
print("Tenants:", tenants)
except Exception as e:
print(f"Error while fetching tenants: {e}")
- refresh\_token: Your temporary Eizen API access token (JWT).
- environment: Which Eizen environment user wants to use dev(AWS) or ldev(local).
- base\_url: Analytical Service URL
- keycloak\_base\_url: Keycloak URL
- client\_id: Keyclock client ID.
- gateway\_base\_url: VIP Gateway Service URL
**To get keys:** Approach the Eizen team to obtain an access\_token for your profile. Note that the access\_token typically has an expiration date.
**Important:** Keep your credentials secure. Do not hardcode them directly in your source code in production environments. Use environment variables or a secure secrets management system.
## **4. Usage Examples**
Here are examples based on your provided code snippets:
### **4.1. Tenant Management (Requires Eizen Admin Privileges)**
- **List All Tenants:**
tenants = sdk.get\_all\_tenants()
print("Available Tenants:")
for tenant in tenants:
` `print(f" ID: {tenant['id']}, Name: {tenant['name']}")
- **Description:** Retrieves a list of all tenants accessible to the authenticated admin user.
- **Returns:** A list of dictionaries, each representing a tenant with details like id and name.
### **4.2. Analytics Type Management (Requires Eizen Admin Privileges)**
- **List All Analytics Types:**
analytics\_types = sdk.get\_all\_analytics\_types()
print("Available Analytics Types:")
for analytic\_type in analytics\_types:
` `print(f" ID: {analytic\_type['id']}, Name: {analytic\_type['name']}")
- **Description:** Retrieves all defined analytics types across tenants.
- **Returns:** A list of dictionaries, each representing an analytics type.
- **Add a New Analytics Type:**
try:
` `new\_type = sdk.add\_analytics\_type(
` `name="Example Analytics Type",
` `description="Description for the new type",
` `tenant\_id=2 # Replace with the target tenant ID
` `)
` `print("Successfully added Analytics Type:")
` `print(new\_type)
except Exception as e:
` `print(f"Error adding analytics type: {e}")
- **Description:** Creates a new analytics type within a specific tenant.
- **Parameters:**
- name (str): The name for the new analytics type.
- description (str): A brief description.
- tenant\_id (int): The ID of the tenant to add this type to.
- **Returns:** A dictionary representing the newly created analytics type.
### **4.3. Analytics Category Management (Requires Eizen Admin Privileges)**
- **List All Analytics Categories:**
analytics\_categories = sdk.get\_all\_analytics\_categories()
print("Available Analytics Categories:")
for category in analytics\_categories:
` `print(f" ID: {category['id']}, Name: {category['name']}")
- **Description:** Retrieves all defined analytics categories.
- **Returns:** A list of dictionaries, each representing an analytics category.
- **Add a New Analytics Category:**
try:
` `new\_category = sdk.add\_analytics\_category(
` `name="Example Category",
` `description="Description for the new category",
` `tenant\_id=2, # Replace with the target tenant ID
` `analytics\_type\_id=51 # Replace with the relevant analytics type ID
` `)
` `print("Successfully added Analytics Category:")
` `print(new\_category)
except Exception as e:
` `print(f"Error adding analytics category: {e}")
- **Description:** Creates a new analytics category under a specific analytics type and tenant.
- **Parameters:**
- name (str): The name for the new category.
- description (str): A brief description.
- tenant\_id (int): The ID of the tenant.
- analytics\_type\_id (int): The ID of the parent analytics type.
- **Returns:** A dictionary representing the newly created analytics category.
### **4.4. Analytics Instance Management (Requires Eizen Admin Privileges)**
- **List All Analytics Instances:**
all\_analytics = sdk.get\_all\_analytics\_list()
print("All Analytics Instances:")
for analytic in all\_analytics:
` `print(f" ID: {analytic['id']}, Name: {analytic['name']}")
- **Description:** Retrieves all analytics instances (specific deployments/configurations).
- **Returns:** A list of dictionaries, each representing an analytics instance.
- **Add a New Analytics Instance:**
try:
` `new\_analytic = sdk.add\_analytics(
` `name="Specific Site Analytics",
` `description="Analytics for a specific site/location",
` `tenant\_id=2, # Replace with the target tenant ID
` `analytics\_type\_id=1, # Replace with the relevant type ID
` `analytics\_category\_id=1 # Replace with the relevant category ID
` `)
` `print("Successfully added Analytics Instance:")
` `print(new\_analytic)
except Exception as e:
` `print(f"Error adding analytics instance: {e}")
- **Description:** Creates a new analytics instance.
- **Parameters:**
- name (str): Name of the instance.
- description (str): Description.
- tenant\_id (int): Owning tenant ID.
- analytics\_type\_id (int): Associated type ID.
- analytics\_category\_id (int): Associated category ID.
- **Returns:** A dictionary representing the newly created analytics instance.
- **Get Analytics Available to User:**
user\_analytics = sdk.get\_analytics()
print("Analytics available to current user:")
# Adjust printing based on the actual structure returned by the SDK
if isinstance(user\_analytics, list):
` `for analytic in user\_analytics:
` `print(f" ID: {analytic.get('id', 'N/A')}, Name: {analytic.get('name', 'N/A')}")
else:
` `print(user\_analytics)
- **Description:** Retrieves the analytics instances the currently authenticated user has access to.
- **Returns:** A list of analytics instances (structure might vary based on SDK implementation).
### **4.5. Zone Management**
- **List All Zones (Accessible to User):**
zones = sdk.get\_all\_zones()
print("All Zones accessible to user:")
# Adjust printing based on the actual structure returned by the SDK
if isinstance(zones, list):
` `for zone in zones:
` `print(f" ID: {zone.get('id', 'N/A')}, Name: {zone.get('name', 'N/A')}")
else:
` `print(zones)
- **Description:** Retrieves all defined zones the user has access to across analytics instances.
- **Returns:** A list of zone dictionaries.
- **Add a New Zone:**
try:
` `# Ensure the user has permission to add zones to this analytic instance
` `analytics\_instance\_id = 46 # Replace with the parent analytics instance ID
` `new\_zone = sdk.add\_zone(
` `name="Specific Area Zone",
` `analytics\_id=analytics\_instance\_id
` `)
` `print("Successfully added Zone:")
` `print(new\_zone)
except Exception as e:
` `print(f"Error adding zone: {e}")
- **Description:** Creates a new zone within a specific analytics instance.
- **Parameters:**
- name (str): Name of the zone.
- analytics\_id (int): The ID of the parent analytics instance.
- **Returns:** A dictionary representing the newly created zone.
- **Get Zones for a Specific Analytic:**
analytic\_id\_to\_query = 25 # Replace with the desired Analytics ID
try:
` `analytic\_zones = sdk.get\_analytic\_zones(analytic\_id=analytic\_id\_to\_query)
` `print(f"Zones for Analytic ID {analytic\_id\_to\_query}:")
` `print(analytic\_zones)
except Exception as e:
` `print(f"Error getting zones for analytic {analytic\_id\_to\_query}: {e}")
- **Description:** Retrieves the zones associated with a specific analytics instance ID (if the user has access).
- **Parameters:**
- analytic\_id (int): The ID of the analytics instance.
- **Returns:** A list or dictionary containing zone information.
### **4.6. Source Management**
- **Get Sources for a Specific Zone:**
zone\_id\_to\_query = 38 # Replace with the desired Zone ID
try:
` `zone\_sources = sdk.get\_zone\_sources(zone\_id=zone\_id\_to\_query)
` `print(f"Sources for Zone ID {zone\_id\_to\_query}:")
` `print(zone\_sources)
except Exception as e:
` `print(f"Error getting sources for zone {zone\_id\_to\_query}: {e}")
- **Description:** Retrieves the sources associated with a specific zone ID (if the user has access).
- **Parameters:**
- zone\_id (int): The ID of the zone.
- **Returns:** A list or dictionary containing source information.
- **Get Sources for a Specific Analytic:**
analytic\_id\_to\_query = 25 # Replace with the desired Analytics ID
try:
` `analytic\_sources = sdk.get\_analytic\_sources(analytic\_id=analytic\_id\_to\_query)
` `print(f"Sources for Analytic ID {analytic\_id\_to\_query}:")
` `print(analytic\_sources)
except Exception as e:
` `print(f"Error getting sources for analytic {analytic\_id\_to\_query}: {e}")
- **Description:** Retrieves all sources associated with a specific analytics instance ID (if the user has access).
- **Parameters:**
- analytic\_id (int): The ID of the analytics instance.
- **Returns:** A list or dictionary containing source information.
- **Get Source Details:**
source\_id\_to\_query = 82 # Replace with the desired Source ID
try:
` `source\_details = sdk.get\_source\_details(source\_id=source\_id\_to\_query)
` `print(f"Details for Source ID {source\_id\_to\_query}:")
` `for key, value in source\_details.items():
` `if key == "models" and isinstance(value, list):
` `print(f" {key}:")
` `for model in value:
` `# Access model details safely using .get()
` `print(f" - Name: {model.get('name', 'N/A')} (ID: {model.get('id', 'N/A')})")
` `else:
` `print(f" {key}: {value}")
except Exception as e:
` `print(f"Error getting details for source {source\_id\_to\_query}: {e}")
- **Description:** Retrieves detailed information about a specific source, including associated models (if the user has access).
- **Parameters:**
- source\_id (int): The ID of the source.
- **Returns:** A dictionary containing detailed source attributes.
- **Get Source Summary:**
source\_id\_to\_query = 82 # Replace with the desired Source ID
try:
` `source\_summary = sdk.get\_source\_summary(source\_id=source\_id\_to\_query)
` `print(f"Summary for Source ID {source\_id\_to\_query}:")
` `print(source\_summary)
except Exception as e:
` `print(f"Error getting summary for source {source\_id\_to\_query}: {e}")
- **Description:** Retrieves summary information or statistics for a specific source (requires backend analytical data and user access).
- **Parameters:**
- source\_id (int): The ID of the source.
- **Returns:** A dictionary containing summary data.
- **Create a New Source (Direct Parameters):**
try:
` `new\_source = sdk.create\_source(
` `name="New Camera Feed Param",
` `username="", # Optional: username for source access
` `password="", # Optional: password for source access
` `source\_url="rtsp://example.com/stream\_param", # URL of the video/data source
` `description="Source created via direct parameters",
` `source\_type="camera", # e.g., camera, video, stream
` `zone\_id=87, # Replace with the target zone ID user has access to
` `# Optional: Mongo details if storing data separately
` `mongo\_host="mongodb://user:pass@host:port", # Replace if needed
` `mongo\_db="database\_name", # Replace if needed
` `models=[116] # List of Model IDs to associate with this source
` `)
` `print("Successfully created Source:")
` `print(new\_source)
except Exception as e:
` `print(f"Error creating source: {e}")
- **Description:** Registers a new data source (like a camera feed) within a zone using direct function arguments.
- **Parameters:** (Include all relevant ones based on the SDK function definition)
- name (str): Name of the source.
- source\_url (str): URL (RTSP, RTMP, HTTP, S3, etc.).
- description (str): Description.
- source\_type (str): Type of source (e.g., camera, video, s3video, image).
- zone\_id (int): Parent zone ID.
- models (list[int]): List of model IDs to run on this source.
- username, password, mongo\_host, mongo\_db (optional, usage depends on source\_type and Eizen configuration).
- **Returns:** A dictionary representing the newly created source.
- **Create Sources from YAML:**
` `# --- Python Code ---
yaml\_file\_path = "/path/to/your/source\_config.yaml" # Make sure this file exists
try:
` `created\_sources\_info = sdk.create\_sources\_from\_yaml(yaml\_file\_path)
` `print("Attempted to create sources from YAML:")
` `print(created\_sources\_info) # Response indicates success/failure per source
except FileNotFoundError:
` `print(f"Error: YAML file not found at {yaml\_file\_path}")
except Exception as e:
` `print(f"Error processing YAML file {yaml\_file\_path}: {e}")
# --- Example `source\_config.yaml` file content ---
"""
# It's recommended to have a top-level key, e.g., 'sources'
# containing a list of source definitions.
# The exact structure depends on how the SDK parses the YAML.
# Assuming the SDK expects a list under 'sources':
- name: "QA Camera 82"
` `username: "" # Optional: username if required by the source URL
` `password: "" # Optional: password if required by the source URL
` `source\_url: "rtmp://10.0.2.185:1935/live/4"
` `description: "QA Source from YAML"
` `source\_type: "camera" # e.g., camera, video, rtsp, s3video
` `zone\_id: 77 # Ensure this Zone ID exists and user has access
` `# Optional: Specify MongoDB if this source requires separate storage
` `mongo\_host: "mongodb://your\_user:your\_password@your\_mongo\_host:27017" # Replace with actual connection string if needed
` `mongo\_db: "eizen\_source\_data" # Replace with actual DB name if needed
` `models: # List of model IDs to run on this source
` `- 50
# Add more source definitions here if needed
# - name: "Another Source"
# ...
"""
- **Description:** Creates one or more sources based on definitions in a YAML file.
- **Parameters:**
- yaml\_file (str): Path to the YAML configuration file.
- **YAML Structure:** The file should contain the source definitions. A common pattern is a top-level key (e.g., sources:) containing a list (-) of dictionaries, where each dictionary defines a source. Adjust the structure based on the SDK's specific requirements.
- **Returns:** Information about the outcome of creating each source defined in the YAML (e.g., a list of results or a summary).
- **Delete a Source:**
source\_id\_to\_delete = 326 # Replace with the ID of the source to delete
try:
` `delete\_result = sdk.delete\_source(source\_id\_to\_delete)
` `print(f"Attempted to delete Source ID {source\_id\_to\_delete}:")
` `# The result might be a confirmation message, status code, or None
` `print(delete\_result)
except Exception as e:
` `print(f"Error deleting source {source\_id\_to\_delete}: {e}")
- **Description:** Deletes a specific source instance. Requires permission to delete the source.
- **Parameters:**
- source\_id (int): The ID of the source to delete.
- **Returns:** A confirmation message or status.
### **4.7. Model Management & Operations**
- **List Available Models:**
try:
` `available\_models = sdk.get\_models()
` `print("Available Models:")
` `# Adjust printing based on the structure returned (likely a list of dicts)
` `if isinstance(available\_models, list):
` `for model in available\_models:
` `print(f" ID: {model.get('id', 'N/A')}, Name: {model.get('name', 'N/A')}, Category: {model.get('category', 'N/A')}")
` `else:
` `print(available\_models)
except Exception as e:
` `print(f"Error retrieving models: {e}")
- **Description:** Retrieves a list of AI models available for use in the platform that the user has access to.
- **Returns:** A list detailing available models.
- **Train (Build) a New Model:**
` `# --- Option 1: Using direct parameters ---
try:
` `training\_job = sdk.model\_building(
` `modelName="My Custom Detector Params",
` `modelCategory="yolo\_detection", # e.g., yolo\_detection, classification, llm\_private\_nas
` `dataPath="s3://eizen-dev/training-data/my\_dataset.zip", # Accessible S3 URL to training data zip
` `numberOfEpochs=50,
` `# Add other relevant parameters like learningRate, batchSize etc. if supported by the SDK function
` `)
` `print("Model training job started (using parameters):")
` `print(training\_job) # Contains job ID or status info
except Exception as e:
` `print(f"Error starting model training (parameters): {e}")
# --- Option 2: Using a YAML file ---
# --- Python Code ---
yaml\_train\_file = "/path/to/your/model\_training.yaml" # Make sure this file exists
try:
` `training\_job\_yaml = sdk.model\_building(yaml\_file=yaml\_train\_file)
` `print(f"Model training job started (using YAML: {yaml\_train\_file}):")
` `print(training\_job\_yaml)
except FileNotFoundError:
` `print(f"Error: YAML training file not found at {yaml\_train\_file}")
except Exception as e:
` `print(f"Error starting model training (YAML): {e}")
# --- Example `model\_training.yaml` file content ---
"""
# Assuming the SDK expects a list under 'training\_config':
training\_config:
` `- modelName: "Elephant Detection YAML"
` `modelCategory: "yolo\_detection" # Must match supported categories
` `# Ensure this S3 URL is accessible by the Eizen training service
` `dataPath: "s3://eizen-dev/elephantbest.zip" # Example from user
` `numberOfEpochs: 10
` `# Add other optional parameters if needed (e.g., learningRate, batchSize)
` `# learningRate: 0.001
` `# You could potentially define multiple training jobs in one YAML
` `# - modelName: "Another Model"
` `# ...
"""
- **Description:** Initiates a new model training process using data stored at dataPath.
- **Parameters (Direct):**
- modelName (str): Name for the new model.
- modelCategory (str): Type of model (must be supported by Eizen).
- dataPath (str): URL (typically S3) to the zipped training dataset. The Eizen platform needs read access to this URL.
- numberOfEpochs (int): Number of training epochs.
- *(Add other optional parameters supported by the function)*
- **Parameters (YAML):**
- yaml\_file (str): Path to the YAML configuration file.
- **YAML Structure:** Similar to sources, likely a top-level key (e.g., training\_config:) containing a list of training job definitions.
- **Returns:** Information about the submitted training job (e.g., job ID, status).
- **Train (Build) a Model Using Custom Weights:**
` `# --- Option 1: Using direct parameters ---
try:
` `training\_job\_custom = sdk.model\_building(
` `modelName="My FineTuned Model Params",
` `modelCategory="llm\_private\_nas", # Example category
` `# URL to custom weights zip (e.g., pre-trained checkpoint)
` `modelWeightsPath="s3://eizen-dev/trainingModelData/safetyweights.zip",
` `useWeights=True, # MUST be True to use modelWeightsPath
` `# dataPath might be needed for fine-tuning data, if applicable
` `# dataPath="s3://my-bucket/finetune-data/data.zip",
` `# numberOfEpochs=20 # Epochs for fine-tuning, if using dataPath
` `)
` `print("Model training job started (using custom weights + params):")
` `print(training\_job\_custom)
except Exception as e:
` `print(f"Error starting model training with custom weights (parameters): {e}")
# --- Option 2: Using a YAML file ---
# --- Python Code ---
yaml\_weights\_file = "/path/to/your/model\_training\_custom\_weights.yaml" # Make sure this file exists
try:
` `training\_job\_yaml\_custom = sdk.model\_building(yaml\_file=yaml\_weights\_file)
` `print(f"Model training job started (using custom weights from YAML: {yaml\_weights\_file}):")
` `print(training\_job\_yaml\_custom)
except FileNotFoundError:
` `print(f"Error: YAML custom weights file not found at {yaml\_weights\_file}")
except Exception as e:
` `print(f"Error starting model training with custom weights (YAML): {e}")
# --- Example `model\_training\_custom\_weights.yaml` file content ---
"""
# Assuming the SDK expects a list under 'training\_config':
training\_config:
` `- modelName: "Safety Detection Model 3 sdk YAML"
` `modelCategory: "llm\_private\_nas" # Must match a supported category
` `# Ensure this S3 URL is accessible by the Eizen training service
` `modelWeightsPath: "s3://eizen-dev/trainingModelData/safetyweights.zip" # User example
` `useWeights: true # Required when providing modelWeightsPath
` `# Add dataPath and numberOfEpochs here if you are fine-tuning
` `# dataPath: "s3://..."
` `# numberOfEpochs: 15
"""
- **Description:** Initiates model training/fine-tuning starting from provided custom weights.
- **Parameters (Direct):**
- modelName, modelCategory.
- modelWeightsPath (str): URL (typically S3) to the custom weights file (e.g., a zip containing model checkpoint/weights). Platform needs read access.
- useWeights (bool): Must be set to True.
- dataPath, numberOfEpochs (optional): Provide if fine-tuning on new data.
- **Parameters (YAML):**
- yaml\_file (str): Path to the YAML file containing the configuration including modelWeightsPath and useWeights: true.
- **Returns:** Information about the submitted training job.
- **Retrain an Existing Model:**
` `# --- Option 1: Using direct parameters ---
try:
` `retraining\_job = sdk.model\_retraining(
` `id=112, # ID of the Eizen model to retrain
` `# URL to the dataset for retraining (platform needs access)
` `dataPath="s3://eizen-dev/trainingModelData-ldev/yolo\_detection/dataset\_retrain.zip",
` `useWeights=True, # Usually True to continue from existing weights
` `numberOfEpochs=30,
` `# Add other relevant retraining parameters if supported
` `)
` `print("Model retraining job started (using parameters):")
` `print(retraining\_job)
except Exception as e:
` `print(f"Error starting model retraining (parameters): {e}")
# --- Option 2: Using a YAML file ---
# --- Python Code ---
yaml\_retrain\_file = "/path/to/your/model\_retraining.yaml" # Make sure this file exists
try:
` `retraining\_job\_yaml = sdk.model\_retraining(yaml\_file=yaml\_retrain\_file)
` `print(f"Model retraining job started (using YAML: {yaml\_retrain\_file}):")
` `print(retraining\_job\_yaml)
except FileNotFoundError:
` `print(f"Error: YAML retraining file not found at {yaml\_retrain\_file}")
except Exception as e:
` `print(f"Error starting model retraining (YAML): {e}")
# --- Example `model\_retraining.yaml` file content ---
"""
# Assuming the SDK expects a list under 'retraining\_config':
retraining\_config:
` `- id: 112 # The existing Eizen Model ID to be retrained
` `# Ensure this S3 URL is accessible by the Eizen training service
` `dataPath: "s3://eizen-dev/trainingModelData-ldev/yolo\_detection/EZA\_MDL\_DETE\_MODE\_TEST/56/data.zip" # User example
` `useWeights: true # Load existing weights before retraining
` `numberOfEpochs: 10 # Number of epochs for this retraining cycle
` `# Add other optional parameters if needed
"""
- **Description:** Initiates a retraining process for an existing model, typically using new data.
- **Parameters (Direct):**
- id (int): The ID of the Eizen model instance to be retrained.
- dataPath (str): URL (typically S3) to the new dataset for retraining. Platform needs access.
- useWeights (bool): Typically True to load the existing model's weights before retraining. False would retrain from scratch using only the new data (less common).
- numberOfEpochs (int): Number of retraining epochs.
- **Parameters (YAML):**
- yaml\_file (str): Path to the YAML configuration file specifying the id and retraining parameters.
- **YAML Structure:** Likely a top-level key (e.g., retraining\_config:) containing a list of retraining job definitions.
- **Returns:** Information about the submitted retraining job.
### **4.8. Inference and Analysis**
- **Perform YOLO Inference on a Video:**
try:
` `inference\_result = sdk.yolo\_inference(
` `model\_id=112, # ID of the trained YOLO model available in Eizen
` `# URL of the input video (e.g., S3, HTTP). Platform needs access.
` `media\_url="s3://cdn-dev.eizen.ai/heatmap\_videos/heatmap\_output\_20250219\_135923.mp4",
` `input\_type="s3video", # Type of input: s3video, httpvideo, rtsp, etc.
` `response\_type="s3videolink" # Desired output: s3videolink, json, annotated\_json etc.
` `# Optional parameters (check SDK function signature):
` `# confidence\_threshold=0.5,
` `# save\_annotated\_video=True,
` `)
` `print("YOLO Video Inference Result:")
` `# The structure of result depends on response\_type
` `print(inference\_result) # e.g., {"output\_url": "s3://...", "job\_id": "..."}
except Exception as e:
` `print(f"Error performing YOLO video inference: {e}")
- **Description:** Runs a specified YOLO model on an input video.
- **Parameters:**
- model\_id (int): The ID of the trained YOLO model in Eizen.
- media\_url (str): URL of the video to process. Eizen platform needs access.
- input\_type (str): Specifies the source/type of the media (s3video, httpvideo, rtsp, etc.).
- response\_type (str): Desired format/location of the result (s3videolink, s3imagelink, json, annotated\_json, callback\_url, etc.).
- **Returns:** Depends on response\_type. Could be a dictionary with a link to the output video/image, JSON results, or a job ID for asynchronous processing.
- **Perform YOLO Inference on an Image:**
` `# Assuming function name is yolo\_inference or similar (verify with SDK)
# If a specific yolo\_infer\_image exists, use that.
try:
` `inference\_result = sdk.yolo\_inference( # Or sdk.yolo\_infer\_image(...)
` `model\_id=72, # ID of the YOLO model
` `# URL of the input image. Platform needs access.
` `media\_url="s3://eizen-dev/videos/test-sdk.png",
` `input\_type="s3image", # Or "s3link", "httpimage" etc.
` `response\_type="s3imagelink" # Or "json", "annotated\_image\_bytes", "annotated\_json" etc.
` `)
` `print("YOLO Image Inference Result:")
` `print(inference\_result) # e.g., {"output\_url": "s3://...", "detections": [...]}
except Exception as e:
` `print(f"Error performing YOLO image inference: {e}")
- **Description:** Runs a specified YOLO model on an input image. *Note: Verify the exact function name in the SDK (yolo\_inference might handle images via input\_type, or a dedicated function like yolo\_infer\_image might exist).*
- **Parameters:** Similar to video inference, but input\_type and response\_type should be image-related.
- **Returns:** Depends on response\_type.
- **Analyze Video with LLaVA (or similar multimodal model):**
video\_url = "s3://cdn-dev.eizen.ai/advertisementVideos/series\_20250109\_172536.mp4" # Platform needs access
instruction = """
See this advertisement video, Analyze this scene and identify the specific show names,
places shown and the actions being performed. Focus on descriptive terms that a
person might search for. Generate up to 3 hashtags, combining both places and
actions. Provide them as a comma-separated list.
\* Examples of Place Hashtags: #CentralPark, #CozyCafe, #MountainView
\* Examples of Action Hashtags: #CookingDinner, #PlayingGuitar, #WalkingDog
Do not include generic terms like 'room', 'location', 'being', 'existing',
'scene', 'video', or 'thing'. Do not repeat similar hashtags.
Only include hashtags directly relevant to what is shown.
"""
try:
` `# Verify actual function name in SDK, e.g., video\_process\_llava, process\_video\_multimodal
` `analysis\_result = sdk.video\_process\_llava(
` `videoUrl=video\_url,
` `instruction=instruction
` `)
` `print("Video Analysis Result (LLaVA):")
` `# Result structure depends on the specific implementation
` `print(analysis\_result) # e.g., {"summary": "...", "hashtags": "#tag1,#tag2", "job\_id": "..."}
except Exception as e:
` `print(f"Error processing video with LLaVA: {e}")
- **Description:** Sends a video and a text prompt/instruction to a multimodal model (like LLaVA) hosted on the Eizen platform for analysis.
- **Parameters:**
- videoUrl (str): URL of the video to analyze. Eizen platform needs access.
- instruction (str): The text prompt guiding the analysis.
- **Returns:** A dictionary containing the model's response (e.g., text summary, extracted information, hashtags) or a job ID.
### **4.9. Data Collection**
- **Collect Data from a Source:**
try:
` `collection\_job = sdk.collect\_data(
` `source\_id=239, # ID of the source to collect data from (must be active)
` `time\_interval=10, # Interval in seconds between captures
` `skip\_frame\_count=5, # Collect 1 frame every 'skip\_frame\_count' + 1 frames (e.g., 5 means 1 of 6)
` `# Optional parameters (check SDK function signature):
` `# duration=3600, # Duration in seconds to collect for (omit for continuous)
` `# target\_path="s3://my-bucket/collected\_data/{source\_id}/" # Optional: Specify S3 destination
` `# target\_format="jpg" # Optional: jpg, png etc.
` `)
` `print("Data collection job started:")
` `# Result likely contains a job ID or confirmation
` `print(collection\_job) # e.g., {"job\_id": "...", "status": "started"}
except Exception as e:
` `print(f"Error starting data collection: {e}")
- **Description:** Initiates a process to capture frames or data segments from a specified, active source. Useful for building datasets for model training/retraining.
- **Parameters:**
- source\_id (int): The source to capture from. Must be an active source like a camera stream.
- time\_interval (int): Approximate seconds between frame grabs.
- skip\_frame\_count (int): Captures 1 frame for every skip\_frame\_count + 1 frames processed from the source. 0 means capture every frame possible subject to time\_interval.
- *(Add other optional parameters like duration, target\_path, target\_format if supported)*
- **Returns:** Information about the submitted data collection job (e.g., job ID, status).
## **5. Error Handling**
SDK methods may raise exceptions on failure (e.g., network issues, invalid parameters, permission denied, API errors). Wrap your SDK calls in try...except blocks to handle potential errors gracefully.
try:
` `# Example: Get details for a source that might not exist or user lacks permission
` `details = sdk.get\_source\_details(source\_id=9999)
` `print(details)
except FileNotFoundError as fnf\_error:
` `print(f"Configuration file error: {fnf\_error}") # Specific error for missing YAML files
except Exception as e:
` `# Log the error, inform the user, or take corrective action
` `print(f"An API call failed: {e}")
` `# Consider inspecting the type of exception or its message for more specific handling
` `# e.g., check for authentication errors, permission errors, etc.
-----